package produce.controller;

import com.alibaba.fastjson.JSON;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.RocketMQTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionState;
import org.springframework.messaging.Message;
import produce.model.Args;

/**
 * 声明监听器的是生产者分组是 "transactional-test" 的 Producer 发送的事务消息。
 * 因为 RocketMQ 是回查（请求）指定指定生产分组下的 Producer，从而获得事务消息的状态，所以一定要正确设置
 */
@RocketMQTransactionListener(txProducerGroup = "transactional-test")
@Slf4j
public class TransactionListenerImpl implements RocketMQLocalTransactionListener {

    /**
     * 实现执行本地事务
     * @param msg
     * @param arg
     * @return
     */
    @Override
    public RocketMQLocalTransactionState executeLocalTransaction(Message msg, Object arg) {
        // 从消息 Header 中解析到 args 参数，并使用 JSON 反序列化
        Args args = JSON.parseObject(msg.getHeaders().get("args", String.class), Args.class);
        // ... local transaction process, return rollback, commit or unknown
        log.info("[executeLocalTransaction][执行本地事务，消息：{} args：{}]", msg, args);
        // Producer 发送事务消息成功，但是因为 executeLocalTransaction 方法返回的是 UNKOWN 状态，所以事务消息并未提交或者回滚
        // RocketMQ Broker 在发送事务消息 30 秒后，发现事务消息还未提交或是回滚，所以回查 Producer 。
        // 此时，checkLocalTransaction 方法返回 COMMIT ，所以该事务消息被提交
        return RocketMQLocalTransactionState.UNKNOWN;
    }

    /**
     * 检查本地事务
     * @param msg
     * @return
     */
    @Override
    public RocketMQLocalTransactionState checkLocalTransaction(Message msg) {
        // ... check transaction status and return rollback, commit or unknown
        log.info("[checkLocalTransaction][回查消息：{}]", msg);
        // 事务消息被提交，所以该消息被 Consumer 消费
        return RocketMQLocalTransactionState.COMMIT;
    }

}